Source code for hysop.operator.spatial_filtering

# Copyright (c) HySoP 2011-2024
#
# This file is part of HySoP software.
# See "https://particle_methods.gricad-pages.univ-grenoble-alpes.fr/hysop-doc/"
# for further info.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.


"""
@file advection.py
RestrictionFilter operator generator.
"""
from hysop.constants import Implementation
from hysop.tools.htypes import check_instance, to_list, first_not_None
from hysop.tools.decorators import debug
from hysop.tools.henum import EnumFactory
from hysop.fields.continuous_field import Field, ScalarField
from hysop.topology.cartesian_descriptor import CartesianTopologyDescriptors
from hysop.parameters.scalar_parameter import ScalarParameter
from hysop.core.graph.computational_node import ComputationalGraphNode
from hysop.core.graph.node_generator import ComputationalGraphNodeGenerator
from hysop.core.graph.computational_node_frontend import (
    MultiComputationalGraphNodeFrontend,
)
from hysop.testsenv import __HAS_OPENCL_BACKEND__

FilteringMethod = EnumFactory.create(
    "FilteringMethod", ["SPECTRAL", "REMESH", "POLYNOMIAL", "SUBGRID"]
)


[docs] class SpatialFilterFrontend(MultiComputationalGraphNodeFrontend): def __new__( cls, input_variable, output_variable, filtering_method, implementation=None, base_kwds=None, **kwds, ): return super().__new__( cls, input_field=None, input_topo=None, output_field=None, output_topo=None, implementation_key=None, implementation=None, base_kwds=base_kwds, **kwds, ) def __init__( self, input_variable, output_variable, filtering_method, implementation=None, base_kwds=None, **kwds, ): """ Initialize a SpatialFilter operator. Parameters ---------- input_variable: ScalarField Input field as a tuple (ScalarField, CartesianTopologyDescriptor). output_variable: ScalarField Output field as a tuple (ScalarField, CartesianTopologyDescriptor). filtering_method: FilteringMethod Specify the lowpass filter type (either spectral or with remeshing kernels). implementation: implementation, optional, defaults to None target implementation, should be contained in available_implementations(). If None, implementation will be set to default_implementation(). base_kwds: dict, optional, defaults to None Base class keywords arguments. If None, an empty dict will be passed. kwds: Extra parameters passed to generated operator. Notes ----- An implementation should at least support the SpatialFilterBase interface. """ check_instance(input_variable, tuple, size=2) check_instance(output_variable, tuple, size=2) check_instance(filtering_method, FilteringMethod) input_field, input_topo = input_variable output_field, output_topo = output_variable check_instance(input_field, ScalarField) check_instance(output_field, ScalarField) check_instance(input_topo, CartesianTopologyDescriptors) check_instance(output_topo, CartesianTopologyDescriptors) check_instance(base_kwds, dict, keys=str, allow_none=True) assert input_topo != output_topo, "Same topology for input and output." super().__init__( input_field=input_field, input_topo=input_topo, output_field=output_field, output_topo=output_topo, implementation_key=filtering_method, implementation=implementation, base_kwds=base_kwds, **kwds, )
[docs] class RestrictionFilterFrontend(SpatialFilterFrontend): """ Interface for restriction filtering: fine grid -> coarse grid Available implementations are: *Python/OpenCL using spectral filtering *Python using remeshing kernels *Python by just taking a subgrid (compatibility with deprecated MultiresolutionFilter) """
[docs] @classmethod def all_implementations(cls): from hysop.backend.host.python.operator.spatial_filtering import ( PythonRemeshRestrictionFilter, PythonSpectralRestrictionFilter, PythonSubgridRestrictionFilter, PythonPolynomialRestrictionFilter, ) ai = { FilteringMethod.SUBGRID: { Implementation.PYTHON: PythonSubgridRestrictionFilter, }, FilteringMethod.POLYNOMIAL: { Implementation.PYTHON: PythonPolynomialRestrictionFilter, }, FilteringMethod.SPECTRAL: { Implementation.PYTHON: PythonSpectralRestrictionFilter, }, FilteringMethod.REMESH: { Implementation.PYTHON: PythonRemeshRestrictionFilter, }, } if __HAS_OPENCL_BACKEND__: from hysop.backend.device.opencl.operator.spatial_filtering import ( OpenClSpectralRestrictionFilter, OpenClSubgridRestrictionFilter, OpenClPolynomialRestrictionFilter, ) ai[FilteringMethod.SUBGRID].update( { Implementation.OPENCL: OpenClSubgridRestrictionFilter, } ) ai[FilteringMethod.POLYNOMIAL].update( { Implementation.OPENCL: OpenClPolynomialRestrictionFilter, } ) ai[FilteringMethod.SPECTRAL].update( { Implementation.OPENCL: OpenClSpectralRestrictionFilter, } ) return ai
[docs] @classmethod def all_default_implementations(cls): adi = { FilteringMethod.SUBGRID: Implementation.PYTHON, FilteringMethod.POLYNOMIAL: Implementation.PYTHON, FilteringMethod.SPECTRAL: Implementation.PYTHON, FilteringMethod.REMESH: Implementation.PYTHON, } return adi
[docs] class InterpolationFilterFrontend(SpatialFilterFrontend): """ Interface for interpolation filtering: coarse grid -> fine grid Available implementations are: *Python using polynomials (linear, cubic, quintic, ...) """
[docs] @classmethod def all_implementations(cls): from hysop.backend.host.python.operator.spatial_filtering import ( PythonPolynomialInterpolationFilter, ) from hysop.backend.device.opencl.operator.spatial_filtering import ( OpenClPolynomialInterpolationFilter, ) ai = { FilteringMethod.POLYNOMIAL: { Implementation.PYTHON: PythonPolynomialInterpolationFilter, Implementation.OPENCL: OpenClPolynomialInterpolationFilter, }, } return ai
[docs] @classmethod def all_default_implementations(cls): adi = { FilteringMethod.POLYNOMIAL: Implementation.PYTHON, } return adi
[docs] class SpatialFilter(ComputationalGraphNodeGenerator): """ Graphnode generator to generate interpolation or restriction filter for multiple fields at once. """ @debug def __new__( cls, input_variables, output_variables, filtering_method, implementation=None, base_kwds=None, **kwds, ): base_kwds = first_not_None(base_kwds, {}) return super().__new__( cls, candidate_input_tensors=None, candidate_output_tensors=None, **base_kwds, ) @debug def __init__( self, input_variables, output_variables, filtering_method, implementation=None, base_kwds=None, **kwds, ): """ Initialize a RestrictionFilter/InterpolationFilter operator generator. Parameters ---------- intput_variables: dict Input fields on fine grid. Dictionary of continuous fields as keys and topologies as values. output_variables: dict Output fields on coarse grid, default to input_fields. Dictionary of continuous fields as keys and topologies as values. filtering_method: FilteringMethod Specify the lowpass filter type (either spectral or with remeshing kernels). implementation: implementation, optional, defaults to None target implementation, should be contained in available_implementations(). If None, implementation will be set to default_implementation(). base_kwds: dict, optional, defaults to None Base class keywords arguments. If None, an empty dict will be passed. kwds: Extra parameters passed to generated operators. """ input_fields = list( ComputationalGraphNode.expand_tensor_fields(input_variables.keys())[0] ) assert len(set(input_fields)) == len(input_fields) output_fields = list( ComputationalGraphNode.expand_tensor_fields(output_variables.keys())[0] ) assert len(input_fields) == len(output_fields) for i, field in enumerate(output_fields): if field is None: output_fields[i] = input_fields[i] input_fields = tuple(input_fields) output_fields = tuple(output_fields) base_kwds = first_not_None(base_kwds, {}) check_instance(input_fields, tuple, values=ScalarField) check_instance(output_fields, tuple, values=ScalarField) check_instance( input_variables, dict, keys=Field, values=CartesianTopologyDescriptors ) check_instance( output_variables, dict, keys=Field, values=CartesianTopologyDescriptors ) check_instance(base_kwds, dict, keys=str) check_instance(filtering_method, FilteringMethod) check_instance(implementation, Implementation) super().__init__( candidate_input_tensors=None, candidate_output_tensors=None, **base_kwds ) self._input_fields = input_fields self._output_fields = output_fields self._input_variables = input_variables self._output_variables = output_variables self._impl = implementation self._fm = filtering_method self._kwds = kwds @debug def _generate(self): nodes = [] for ifield, ofield in zip(self._input_fields, self._output_fields): stopo = ComputationalGraphNode.get_topo_discretization( self._input_variables, ifield ) ttopo = ComputationalGraphNode.get_topo_discretization( self._output_variables, ofield ) check_instance(stopo, tuple, values=int) check_instance(ttopo, tuple, values=int) assert len(stopo) == len(ttopo) fm = self._fm impl = self._impl kwds = self._kwds.copy() # if source topology is destination topology there is nothing to be done if ttopo == stopo: continue elif all(ns <= nt for (ns, nt) in zip(stopo, ttopo)): # here we build an interpolation filter operator node = InterpolationFilterFrontend( input_variable=(ifield, stopo), output_variable=(ofield, ttopo), filtering_method=fm, implementation=impl, **kwds, ) elif all(ns >= nt for (ns, nt) in zip(stopo, ttopo)): # here we build a restriction filter operator node = RestrictionFilterFrontend( input_variable=(ifield, stopo), output_variable=(ofield, ttopo), filtering_method=fm, implementation=impl, **kwds, ) else: msg = ( "Inconsistant topology descriptors {} and {} for field {} and {}, " ) msg += "cannot interpolate and restrict at the same time." msg = msg.format(stopo, ttopo, ifield.name, ofield.name) raise RuntimeError(msg) nodes.append(node) return nodes